package com.zyguo.sdk.websocket;

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.HashMap;

import org.apache.log4j.Logger;
import org.eclipse.jetty.util.thread.QueuedThreadPool;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
import org.eclipse.jetty.websocket.api.annotations.WebSocket;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.eclipse.jetty.websocket.client.WebSocketClient;

import com.zyguo.sdk.utils.InputStream2Byte;

/**
 * Basic Echo Client Socket
 */
@WebSocket(maxTextMessageSize = 64 * 1024)
public class WSClient<T>{
	private Logger logger = Logger.getLogger(WSClient.class);
	private Session session;
	private String name = null;
	private WebSocketClient client = null;
	private URI wsUri = null;
	private HashMap<Long, ResponseLock> lockMap = new HashMap<Long, ResponseLock>();
	private int responseWaitTime = 1000 ;
	private boolean isHealth = false;
	
	public WSClient(String uri, String name, int responseWaitTime  ) {
		this.responseWaitTime = responseWaitTime;
		this.name = name;
		QueuedThreadPool threadPool = new QueuedThreadPool();
		threadPool.setName( name );
		threadPool.setDaemon( false );
		client = new WebSocketClient( threadPool );
		try {
			wsUri = new URI( uri );
			connect();
		} catch (URISyntaxException e) {
			logger.error("uri=" + uri +",有问题", e);
		}
		
	}

	public String getName() {
		return name;
	}

	@OnWebSocketClose
	public void onClose(int statusCode, String reason) {
		logger.debug(String.format("Connection closed: %d - %s%n", statusCode, reason) + ",threadName=" + this.name );
		this.session = null;
		this.isHealth = false;
	}

	@OnWebSocketConnect
	public void onConnect(Session session) throws InterruptedException,	IOException {
		logger.debug(String.format("Got connect: %s%n", session) + ",threadName=" + this.name );
		this.session = session;
		this.isHealth = true;
	}

	
	@OnWebSocketMessage
	public void onMessage(InputStream is){
		logger.debug("onMessage, threadName="  + this.name + ",is=" + is );
		try {
			// 处理消息
			byte[] outputbypes = InputStream2Byte.inputStreamToByte(is); 
			ResponseMsg resp = NetTransferUtils.getResponseMsg( outputbypes );
			logger.debug("onMessage, resp="  + resp );
			if( resp != null ){
				ResponseLock lock =  lockMap.get( resp.getMsgId() );
				if( lock != null ){
					lock.setResponseMsg( resp );	
				}
				//dealResponse( resp );
			}
			logger.debug("onMessage, after notify, threadName="  + this.name );
			
		} catch (Exception e) {
			logger.error("onMessage, 处理 异常" ,e );
		}
		
	}
	


	public ResponseMsg sendMsgByAsyn( RequestMsg<T> msg ){
		logger.debug("sendMsgByAsyn, threadName="  + this.name +",msg=" + msg );
		if ( msg != null) {
			byte[] inputbytes = NetTransferUtils.getRequestMsg( msg );
			ByteBuffer buffer = ByteBuffer.wrap(inputbytes);
			try {
				session.getRemote().sendBytes(buffer);
			} catch (IOException e) {
				isHealth = false;
				logger.error("发送消息失败,msg=" + msg.getHeader() + e );
			}
			logger.debug("send Msg,t="  + System.currentTimeMillis() + ",header=" + msg.getHeader() );
			ResponseLock resLock = new ResponseLock( responseWaitTime );
			lockMap.put( msg.getHeader().getMsgId(), resLock ); 
			ResponseMsg responseMsg = resLock.getResponseMsg();
			//需要删除掉,否则就只增不减了，容易内存溢出了
			lockMap.remove( resLock );
			logger.debug("sendMsg, after await, t=" + System.currentTimeMillis() + ",response=" + responseMsg + ",header=" + msg.getHeader() );
			//System.out.println("sendMsg, after await, t=" + System.currentTimeMillis() + "response=" + responseMsg + ",header=" + msg.getHeader() );
			return responseMsg;
		} 
		return null;
		
	}

	public boolean sendMsgBySync( RequestMsg<T> msg ){
		logger.debug("sendMsgBySync, threadName="  + this.name +",msg=" + msg );
		if ( msg != null) {
			byte[] inputbytes = NetTransferUtils.getRequestMsg( msg );
			ByteBuffer buffer = ByteBuffer.wrap(inputbytes);
			try {
				session.getRemote().sendBytes(buffer);
				logger.debug("send Msg,t="  + System.currentTimeMillis() + ",header=" + msg.getHeader() );
				return true;
			} catch (IOException e) {
				isHealth = false;
				logger.error("发送消息失败,msg=" + msg.getHeader() + e );
			}
		} 
		return false;
		
	}
	
	private void connect(){
		try {
			logger.debug("尝试连接 wsuri=" + wsUri + ",threadName=" + this.name );
			ClientUpgradeRequest request = new ClientUpgradeRequest();
			client.start();
			client.connect(this, wsUri, request);
			client.setMaxIdleTimeout(Long.MAX_VALUE);
			// socket.awaitClose(5, TimeUnit.SECONDS);
		} catch (Throwable t) {
			logger.error("connect error:", t);
			isHealth = false;
		}
	}
	
	public boolean isHealth() {
		return isHealth;
	}

	public void reconnect() {
		this.connect();
	}
}

